package com.biglead.orderservice.kafka;


import com.alibaba.fastjson.JSON;
import com.biglead.common.pojo.OrderMessage;
import com.biglead.orderservice.service.OrderKafakaService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;


@Slf4j
@Component
public class OrderConsumerListen {


    @Resource
    private OrderKafakaService orderKafakaService;

    @KafkaListener(topics = "ORDER-CREATE-TOPIC")
    public void listen(ConsumerRecord<String, String> record) throws Exception {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        // Object -> String
        String message = (String) kafkaMessage.get();
        // 反序列化
        OrderMessage orderMessage = JSON.parseObject(message,OrderMessage.class);
        // 创建订单
        orderKafakaService.consumerTopicToCreateOrderWithKafka(orderMessage);
    }
}